package com.boyi.center.task;

import com.boyi.shared.kafka.KafkaResult;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

/**
 * @author tingwen
 */

public abstract class AbstractKafkaTask implements KafkaTask {
    @Autowired
    private KafkaTemplate kafkaTemplate;

    final static int FAILURE_TIME = 5;

    @Override
    public void taskPublic(ProducerRecord<String, KafkaResult<Object>> record) {
        ListenableFuture future = kafkaTemplate.send(record);
        future.addCallback(new ListenableFutureCallback() {
            @Override
            public void onFailure(Throwable ex) {
                handleFailure(record);
            }

            @Override
            public void onSuccess(Object result) {
                handleSuccess(result);
            }
        });
    }
}
